/*-------------------------------------------------------------------------
 *
 * publicationcmds.c
 *		发布操作
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *		src/backend/commands/publicationcmds.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_type.h"
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
#include "parser/parse_collate.h"
#include "parser/parse_relation.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/varlena.h"


/*
 * 用于验证行过滤器表达式中列的信息。有关详细信息，请参见
 * contain_invalid_rfcolumn_walker。
 */
typedef struct rf_context
{
	Bitmapset  *bms_replident;	/* 副本身份列的位集合 */
	bool		pubviaroot;		/* 如果我们正在验证父关系的行过滤器则为真 */
	Oid			relid;			/* 关系的 relid */
	Oid			parentid;		/* 父关系的 relid */
} rf_context;

static List *fc_OpenTableList(List *fc_tables);
static void fc_CloseTableList(List *fc_rels);
static void fc_LockSchemaList(List *fc_schemalist);
static void fc_PublicationAddTables(Oid fc_pubid, List *fc_rels, bool fc_if_not_exists,
								 AlterPublicationStmt *fc_stmt);
static void fc_PublicationDropTables(Oid fc_pubid, List *fc_rels, bool fc_missing_ok);
static void fc_PublicationAddSchemas(Oid fc_pubid, List *fc_schemas, bool fc_if_not_exists,
								  AlterPublicationStmt *fc_stmt);
static void fc_PublicationDropSchemas(Oid fc_pubid, List *fc_schemas, bool fc_missing_ok);


static void fc_parse_publication_options(ParseState *fc_pstate,
						  List *fc_options,
						  bool *fc_publish_given,
						  PublicationActions *fc_pubactions,
						  bool *fc_publish_via_partition_root_given,
						  bool *fc_publish_via_partition_root)
{
	ListCell   *fc_lc;

	*fc_publish_given = false;
	*fc_publish_via_partition_root_given = false;

	/* 默认值 */
	fc_pubactions->pubinsert = true;
	fc_pubactions->pubupdate = true;
	fc_pubactions->pubdelete = true;
	fc_pubactions->pubtruncate = true;
	*fc_publish_via_partition_root = false;

	/* 解析选项 */
	foreach(fc_lc, fc_options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_lc);

		if (strcmp(fc_defel->defname, "publish") == 0)
		{
			char	   *fc_publish;
			List	   *fc_publish_list;
			ListCell   *fc_lc;

			if (*fc_publish_given)
				errorConflictingDefElem(fc_defel, fc_pstate);

			/*
			 * 如果提供了发布选项，则只应发布明确列出的操作。
			 */
			fc_pubactions->pubinsert = false;
			fc_pubactions->pubupdate = false;
			fc_pubactions->pubdelete = false;
			fc_pubactions->pubtruncate = false;

			*fc_publish_given = true;
			fc_publish = defGetString(fc_defel);

			if (!SplitIdentifierString(fc_publish, ',', &fc_publish_list))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("invalid list syntax in parameter \"%s\"",
								"publish")));

			/* 处理选项列表。 */
			foreach(fc_lc, fc_publish_list)
			{
				char	   *fc_publish_opt = (char *) lfirst(fc_lc);

				if (strcmp(fc_publish_opt, "insert") == 0)
					fc_pubactions->pubinsert = true;
				else if (strcmp(fc_publish_opt, "update") == 0)
					fc_pubactions->pubupdate = true;
				else if (strcmp(fc_publish_opt, "delete") == 0)
					fc_pubactions->pubdelete = true;
				else if (strcmp(fc_publish_opt, "truncate") == 0)
					fc_pubactions->pubtruncate = true;
				else
					ereport(ERROR,
							(errcode(ERRCODE_SYNTAX_ERROR),
							 errmsg("unrecognized value for publication option \"%s\": \"%s\"",
									"publish", fc_publish_opt)));
			}
		}
		else if (strcmp(fc_defel->defname, "publish_via_partition_root") == 0)
		{
			if (*fc_publish_via_partition_root_given)
				errorConflictingDefElem(fc_defel, fc_pstate);
			*fc_publish_via_partition_root_given = true;
			*fc_publish_via_partition_root = defGetBoolean(fc_defel);
		}
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("unrecognized publication parameter: \"%s\"", fc_defel->defname)));
	}
}

/*
 * 将 PublicationObjSpecType 列表转换为模式 oid 列表和
 * PublicationTable 列表。
 */
static void fc_ObjectsInPublicationToOids(List *fc_pubobjspec_list, ParseState *fc_pstate,
						   List **fc_rels, List **fc_schemas)
{
	ListCell   *fc_cell;
	PublicationObjSpec *fc_pubobj;

	if (!fc_pubobjspec_list)
		return;

	foreach(fc_cell, fc_pubobjspec_list)
	{
		Oid			fc_schemaid;
		List	   *fc_search_path;

		fc_pubobj = (PublicationObjSpec *) lfirst(fc_cell);

		switch (fc_pubobj->pubobjtype)
		{
			case PUBLICATIONOBJ_TABLE:
				*fc_rels = lappend(*fc_rels, fc_pubobj->pubtable);
				break;
			case PUBLICATIONOBJ_TABLES_IN_SCHEMA:
				fc_schemaid = get_namespace_oid(fc_pubobj->name, false);

				/* 如果用户指定“sch1, sch1”，则过滤掉重复项 */
				*fc_schemas = list_append_unique_oid(*fc_schemas, fc_schemaid);
				break;
			case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA:
				fc_search_path = fetch_search_path(false);
				if (fc_search_path == NIL) /* search_path 中没有有效项？ */
					ereport(ERROR,
							errcode(ERRCODE_UNDEFINED_SCHEMA),
							errmsg("no schema has been selected for CURRENT_SCHEMA"));

				fc_schemaid = linitial_oid(fc_search_path);
				list_free(fc_search_path);

				/* 如果用户指定“sch1, sch1”，则过滤掉重复项 */
				*fc_schemas = list_append_unique_oid(*fc_schemas, fc_schemaid);
				break;
			default:
				/* 不应该发生 */
				elog(ERROR, "invalid publication object type %d", fc_pubobj->pubobjtype);
				break;
		}
	}
}

/*
 * 如果行过滤器 WHERE 表达式中使用的任意列不是
 * REPLICA IDENTITY 的一部分，则返回真，否则返回假。
 */
static bool fc_contain_invalid_rfcolumn_walker(Node *fc_node, rf_context *fc_context)
{
	if (fc_node == NULL)
		return false;

	if (IsA(fc_node, Var))
	{
		Var		   *fc_var = (Var *) fc_node;
		AttrNumber	fc_attnum = fc_var->varattno;

		/*
		 * 如果 pubviaroot 为真，我们正在验证父表的行过滤器，
		 * 但位图包含子表的副本身份
		 * 信息。因此，获取子表的列编号，因为父子列的顺序可能不同。
		 */
		if (fc_context->pubviaroot)
		{
			char	   *fc_colname = get_attname(fc_context->parentid, fc_attnum, false);

			fc_attnum = get_attnum(fc_context->relid, fc_colname);
		}

		if (!bms_is_member(fc_attnum - FirstLowInvalidHeapAttributeNumber,
						   fc_context->bms_replident))
			return true;
	}

	return expression_tree_walker(fc_node, fc_contain_invalid_rfcolumn_walker,
								  (void *) fc_context);
}

/*
 * 检查过滤表达式中引用的所有列是否为
 * REPLICA IDENTITY 索引的一部分。
 *
 * 如果找到任何无效列，则返回真。
 */
bool pub_rf_contains_invalid_column(Oid fc_pubid, Relation fc_relation, List *fc_ancestors,
							   bool fc_pubviaroot)
{
	HeapTuple	fc_rftuple;
	Oid			fc_relid = RelationGetRelid(fc_relation);
	Oid			fc_publish_as_relid = RelationGetRelid(fc_relation);
	bool		fc_result = false;
	Datum		fc_rfdatum;
	bool		fc_rfisnull;

	/*
	 * FULL 表示所有列都在 REPLICA IDENTITY 中，因此所有列均
	 * 可在行过滤器中使用，我们可以跳过验证。
	 */
	if (fc_relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
		return false;

	/*
	 * 对于一个分区，如果 pubviaroot 为真，找到通过此发布
	 * 发布的最顶层祖先，因为我们需要使用其行过滤器
	 * 表达式来过滤分区的更改。
	 *
	 * 请注意，尽管使用的行过滤器是针对祖先的，但使用的
	 * REPLICA IDENTITY 将是实际子表的。
	 */
	if (fc_pubviaroot && fc_relation->rd_rel->relispartition)
	{
		fc_publish_as_relid
			= GetTopMostAncestorInPublication(fc_pubid, fc_ancestors, NULL);

		if (!OidIsValid(fc_publish_as_relid))
			fc_publish_as_relid = fc_relid;
	}

	fc_rftuple = SearchSysCache2(PUBLICATIONRELMAP,
							  ObjectIdGetDatum(fc_publish_as_relid),
							  ObjectIdGetDatum(fc_pubid));

	if (!HeapTupleIsValid(fc_rftuple))
		return false;

	fc_rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_rftuple,
							  Anum_pg_publication_rel_prqual,
							  &fc_rfisnull);

	if (!fc_rfisnull)
	{
		rf_context	fc_context = {0};
		Node	   *fc_rfnode;
		Bitmapset  *fc_bms = NULL;

		fc_context.pubviaroot = fc_pubviaroot;
		fc_context.parentid = fc_publish_as_relid;
		fc_context.relid = fc_relid;

		/* 记住作为 REPLICA IDENTITY 的列 */
		fc_bms = RelationGetIndexAttrBitmap(fc_relation,
										 INDEX_ATTR_BITMAP_IDENTITY_KEY);

		fc_context.bms_replident = fc_bms;
		fc_rfnode = stringToNode(TextDatumGetCString(fc_rfdatum));
		fc_result = fc_contain_invalid_rfcolumn_walker(fc_rfnode, &fc_context);
	}

	ReleaseSysCache(fc_rftuple);

	return fc_result;
}

/*
 * 检查 REPLICA IDENTITY 中引用的所有列是否被
 * 列表覆盖。
 *
 * 如果任何副本身份列未被列名单覆盖，则返回真。
 */
bool pub_collist_contains_invalid_column(Oid fc_pubid, Relation fc_relation, List *fc_ancestors,
									bool fc_pubviaroot)
{
	HeapTuple	fc_tuple;
	Oid			fc_relid = RelationGetRelid(fc_relation);
	Oid			fc_publish_as_relid = RelationGetRelid(fc_relation);
	bool		fc_result = false;
	Datum		fc_datum;
	bool		fc_isnull;

	/*
	 * 对于一个分区，如果 pubviaroot 为真，找到通过此发布
	 * 发布的最顶层祖先，因为我们需要使用其列列表来处理
	 * 更改。
	 *
	 * 请注意，尽管使用的列列表是针对祖先的，但使用的
	 * REPLICA IDENTITY 将是实际子表的。
	 */
	if (fc_pubviaroot && fc_relation->rd_rel->relispartition)
	{
		fc_publish_as_relid = GetTopMostAncestorInPublication(fc_pubid, fc_ancestors, NULL);

		if (!OidIsValid(fc_publish_as_relid))
			fc_publish_as_relid = fc_relid;
	}

	fc_tuple = SearchSysCache2(PUBLICATIONRELMAP,
							ObjectIdGetDatum(fc_publish_as_relid),
							ObjectIdGetDatum(fc_pubid));

	if (!HeapTupleIsValid(fc_tuple))
		return false;

	fc_datum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_tuple,
							Anum_pg_publication_rel_prattrs,
							&fc_isnull);

	if (!fc_isnull)
	{
		int			fc_x;
		Bitmapset  *fc_idattrs;
		Bitmapset  *fc_columns = NULL;

		/* 使用 REPLICA IDENTITY FULL，不允许列列表。 */
		if (fc_relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
			fc_result = true;

		/* 将列列表数据转换为位图集。 */
		fc_columns = pub_collist_to_bitmapset(NULL, fc_datum, NULL);

		/* 记住作为 REPLICA IDENTITY 的列 */
		fc_idattrs = RelationGetIndexAttrBitmap(fc_relation,
											 INDEX_ATTR_BITMAP_IDENTITY_KEY);

		/*
		 * 由 RelationGetIndexAttrBitmap 返回的位图中的 Attnums
		 * 是偏移的（以处理系统列的常规方式），而列列表
		 * 不使用偏移，因此我们无法使用 bms_is_subset()。相反，我们
		 * 必须循环遍历 idattrs 并检查它们全部是否在
		 * 列表中。
		 */
		fc_x = -1;
		while ((fc_x = bms_next_member(fc_idattrs, fc_x)) >= 0)
		{
			AttrNumber	fc_attnum = (fc_x + FirstLowInvalidHeapAttributeNumber);

			/*
			 * 如果 pubviaroot 为真，我们正在验证父表的列列表，
			 * 但位图包含子表的副本标识信息。父/子 attnums 可能不匹配，
			 * 因此需要将它们转换为父级 - 从子级获取 attname，
			 * 并在父级中查找它。
			 */
			if (fc_pubviaroot)
			{
				/* 子表中的属性名称 */
				char	   *fc_colname = get_attname(fc_relid, fc_attnum, false);

				/*
				 * 确定父级中属性名称的 attnum（我们正在使用
				 * 定义在父级上的列列表）。
				 */
				fc_attnum = get_attnum(fc_publish_as_relid, fc_colname);
			}

			/* 副本标识列，不在列列表中 */
			if (!bms_is_member(fc_attnum, fc_columns))
			{
				fc_result = true;
				break;
			}
		}

		bms_free(fc_idattrs);
		bms_free(fc_columns);
	}

	ReleaseSysCache(fc_tuple);

	return fc_result;
}

/* check_functions_in_node 回调 */
static bool fc_contain_mutable_or_user_functions_checker(Oid fc_func_id, void *fc_context)
{
	return (func_volatile(fc_func_id) != PROVOLATILE_IMMUTABLE ||
			fc_func_id >= FirstNormalObjectId);
}

/*
 * 行过滤器 walker 检查行过滤器表达式是否是“简单
 * 表达式”。
 *
 * 它只允许简单或复合表达式，例如：
 * - (Var Op Const)
 * - (Var Op Var)
 * - (Var Op Const) AND/OR (Var Op Const)
 * - 等等
 * （其中 Var 是该过滤器所属表的列）
 *
 * 简单表达式有以下限制：
 * - 不允许用户定义的运算符；
 * - 不允许用户定义的函数；
 * - 不允许用户定义的类型；
 * - 不允许用户定义的排序规则；
 * - 不允许非不可变的内置函数；
 * - 不允许系统列。
 *
 * 注意事项
 *
 * 我们不允许用户定义的函数/运算符/类型/排序规则，因为
 * (a) 如果用户删除了在行过滤器表达式中使用的用户定义对象，
 * 或如果在使用它时出现其他错误，逻辑解码
 * 基础设施将无法从这种错误中恢复，即使该对象被重新创建，
 * 因为使用的是历史快照来评估行过滤器；
 * (b) 用户定义的函数可用于访问可能导致不良结果的表，
 * 因为使用的是历史快照。这就是为什么在行过滤器表达式中
 * 只允许不可变的内置函数。
 *
 * 我们不允许系统列，因为目前我们没有在传递给下游的元组中
 * 得到该信息。此外，由于我们不将这些列复制到订阅者，
 * 因此似乎没有必要对这些列进行过滤。
 *
 * 我们可以在更多分析和测试后允许其他节点类型。
 */
static bool fc_check_simple_rowfilter_expr_walker(Node *fc_node, ParseState *fc_pstate)
{
	char	   *fc_errdetail_msg = NULL;

	if (fc_node == NULL)
		return false;

	switch (nodeTag(fc_node))
	{
		case T_Var:
			/* 不允许系统列。 */
			if (((Var *) fc_node)->varattno < InvalidAttrNumber)
				fc_errdetail_msg = _("System columns are not allowed.");
			break;
		case T_OpExpr:
		case T_DistinctExpr:
		case T_NullIfExpr:
			/* 可以，但不允许用户定义的运算符。 */
			if (((OpExpr *) fc_node)->opno >= FirstNormalObjectId)
				fc_errdetail_msg = _("User-defined operators are not allowed.");
			break;
		case T_ScalarArrayOpExpr:
			/* 可以，但不允许用户定义的运算符。 */
			if (((ScalarArrayOpExpr *) fc_node)->opno >= FirstNormalObjectId)
				fc_errdetail_msg = _("User-defined operators are not allowed.");

			/*
			 * 我们不需要检查 ScalarArrayOpExpr 的 hashfuncid 和 negfuncid
			 * 因为这些函数仅针对子查询构建。
			 */
			break;
		case T_RowCompareExpr:
			{
				ListCell   *fc_opid;

				/* 可以，但不允许用户定义的运算符。 */
				foreach(fc_opid, ((RowCompareExpr *) fc_node)->opnos)
				{
					if (lfirst_oid(fc_opid) >= FirstNormalObjectId)
					{
						fc_errdetail_msg = _("User-defined operators are not allowed.");
						break;
					}
				}
			}
			break;
		case T_Const:
		case T_FuncExpr:
		case T_BoolExpr:
		case T_RelabelType:
		case T_CollateExpr:
		case T_CaseExpr:
		case T_CaseTestExpr:
		case T_ArrayExpr:
		case T_RowExpr:
		case T_CoalesceExpr:
		case T_MinMaxExpr:
		case T_XmlExpr:
		case T_NullTest:
		case T_BooleanTest:
		case T_List:
			/* 可以，支持 */
			break;
		default:
			fc_errdetail_msg = _("Only columns, constants, built-in operators, built-in data types, built-in collations, and immutable built-in functions are allowed.");
			break;
	}

	/*
	 * 对于所有支持的节点，如果我们还没有发现问题，
	 * 检查其中使用的类型、函数和排序规则。我们通过遍历
	 * 每个元素来检查 List。
	 */
	if (!fc_errdetail_msg && !IsA(fc_node, List))
	{
		if (exprType(fc_node) >= FirstNormalObjectId)
			fc_errdetail_msg = _("User-defined types are not allowed.");
		else if (check_functions_in_node(fc_node, fc_contain_mutable_or_user_functions_checker,
										 (void *) fc_pstate))
			fc_errdetail_msg = _("User-defined or built-in mutable functions are not allowed.");
		else if (exprCollation(fc_node) >= FirstNormalObjectId ||
				 exprInputCollation(fc_node) >= FirstNormalObjectId)
			fc_errdetail_msg = _("User-defined collations are not allowed.");
	}

	/*
	 * 如果我们在此节点中发现了问题，请立即抛出错误。
	 * 否则继续进行。
	 */
	if (fc_errdetail_msg)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("invalid publication WHERE expression"),
				 errdetail_internal("%s", fc_errdetail_msg),
				 parser_errposition(fc_pstate, exprLocation(fc_node))));

	return expression_tree_walker(fc_node, fc_check_simple_rowfilter_expr_walker,
								  (void *) fc_pstate);
}

/*
 * 检查行过滤表达式是否为“简单表达式”。
 *
 * 详情请参见 check_simple_rowfilter_expr_walker。
 */
static bool fc_check_simple_rowfilter_expr(Node *fc_node, ParseState *fc_pstate)
{
	return fc_check_simple_rowfilter_expr_walker(fc_node, fc_pstate);
}

/*
 * 转换列表中所有关系的发布 WHERE 表达式，确保它被强制转换为布尔值，如果需要，则添加必要的排序信息，并将与该关系相关的新 nsitem/RTE 添加到 ParseState 的命名空间列表中。
 *
 * 同时检查发布行过滤表达式，如果遇到任何不允许或意外的情况则抛出错误。
 */
static void fc_TransformPubWhereClauses(List *fc_tables, const char *fc_queryString,
						 bool fc_pubviaroot)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_tables)
	{
		ParseNamespaceItem *fc_nsitem;
		Node	   *fc_whereclause = NULL;
		ParseState *fc_pstate;
		PublicationRelInfo *fc_pri = (PublicationRelInfo *) lfirst(fc_lc);

		if (fc_pri->whereClause == NULL)
			continue;

		/*
		 * 如果发布不通过根分区表发布更改，则将使用分区的行过滤器。因此在这种情况下不允许在分区表上使用 WHERE 子句。
		 */
		if (!fc_pubviaroot &&
			fc_pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("cannot use publication WHERE clause for relation \"%s\"",
							RelationGetRelationName(fc_pri->relation)),
					 errdetail("WHERE clause cannot be used for a partitioned table when %s is false.",
							   "publish_via_partition_root")));

		/*
		 * 需要一个新的 pstate，以便在其范围表中只有“这个”表
		 */
		fc_pstate = make_parsestate(NULL);
		fc_pstate->p_sourcetext = fc_queryString;
		fc_nsitem = addRangeTableEntryForRelation(fc_pstate, fc_pri->relation,
											   AccessShareLock, NULL,
											   false, false);
		addNSItemToQuery(fc_pstate, fc_nsitem, false, true, true);

		fc_whereclause = transformWhereClause(fc_pstate,
										   copyObject(fc_pri->whereClause),
										   EXPR_KIND_WHERE,
										   "PUBLICATION WHERE");

		/* 修复排序信息 */
		assign_expr_collations(fc_pstate, fc_whereclause);

		/*
		 * 我们仅允许在行过滤器中使用简单表达式。详见 check_simple_rowfilter_expr_walker。
		 */
		fc_check_simple_rowfilter_expr(fc_whereclause, fc_pstate);

		free_parsestate(fc_pstate);

		fc_pri->whereClause = fc_whereclause;
	}
}


/*
 * 给定即将添加到发布的表列表，验证它们是否满足必要的前提条件，即：如果任何模式被发布，则没有表具有列列表；如果未设置 publish_via_partition_root，则分区表没有列列表。
 *
 * 'publish_schema' 表示发布包含任何 TABLES IN SCHEMA 元素（在此命令中新添加或已存在）。
 * 'pubviaroot' 是 publish_via_partition_root 的值。
 */
static void fc_CheckPubRelationColumnList(char *fc_pubname, List *fc_tables,
						   bool fc_publish_schema, bool fc_pubviaroot)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_tables)
	{
		PublicationRelInfo *fc_pri = (PublicationRelInfo *) lfirst(fc_lc);

		if (fc_pri->columns == NIL)
			continue;

		/*
		 * 如果任何模式在发布中，则不允许指定列列表。
		 *
		 * XXX 我们可以禁止发布尝试发布该表具有列列表及模式的情况。然而，如果我们这样做，则在 ALTER TABLE ... SET SCHEMA 期间需要一个限制，以防止此类情况，这似乎不是一个好主意。
		 */
		if (fc_publish_schema)
			ereport(ERROR,
					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
						   get_namespace_name(RelationGetNamespace(fc_pri->relation)),
						   RelationGetRelationName(fc_pri->relation), fc_pubname),
					errdetail("Column lists cannot be specified in publications containing FOR TABLES IN SCHEMA elements."));

		/*
		 * 如果发布不通过根分区表发布更改，则将使用分区的列列表。因此在这种情况下不允许在分区表上使用列列表。
		 */
		if (!fc_pubviaroot &&
			fc_pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
							get_namespace_name(RelationGetNamespace(fc_pri->relation)),
							RelationGetRelationName(fc_pri->relation), fc_pubname),
					 errdetail("Column lists cannot be specified for partitioned tables when %s is false.",
							   "publish_via_partition_root")));
	}
}

/*
 * 创建新发布。
 */
ObjectAddress CreatePublication(ParseState *fc_pstate, CreatePublicationStmt *fc_stmt)
{
	Relation	fc_rel;
	ObjectAddress fc_myself;
	Oid			fc_puboid;
	bool		fc_nulls[Natts_pg_publication];
	Datum		fc_values[Natts_pg_publication];
	HeapTuple	fc_tup;
	bool		fc_publish_given;
	PublicationActions fc_pubactions;
	bool		fc_publish_via_partition_root_given;
	bool		fc_publish_via_partition_root;
	AclResult	fc_aclresult;
	List	   *fc_relations = NIL;
	List	   *fc_schemaidlist = NIL;

	/* 必须对数据库具有 CREATE 权限 */
	fc_aclresult = pg_database_aclcheck(MyDatabaseId, GetUserId(), ACL_CREATE);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, OBJECT_DATABASE,
					   get_database_name(MyDatabaseId));

	/* 对所有表的要求是超级用户 */
	if (fc_stmt->for_all_tables && !superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to create FOR ALL TABLES publication")));

	fc_rel = table_open(PublicationRelationId, RowExclusiveLock);

	/* 检查名称是否已使用 */
	fc_puboid = GetSysCacheOid1(PUBLICATIONNAME, Anum_pg_publication_oid,
							 CStringGetDatum(fc_stmt->pubname));
	if (OidIsValid(fc_puboid))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("publication \"%s\" already exists",
						fc_stmt->pubname)));

	/* 形成一个元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));

	fc_values[Anum_pg_publication_pubname - 1] =
		DirectFunctionCall1(namein, CStringGetDatum(fc_stmt->pubname));
	fc_values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId());

	fc_parse_publication_options(fc_pstate,
							  fc_stmt->options,
							  &fc_publish_given, &fc_pubactions,
							  &fc_publish_via_partition_root_given,
							  &fc_publish_via_partition_root);

	fc_puboid = GetNewOidWithIndex(fc_rel, PublicationObjectIndexId,
								Anum_pg_publication_oid);
	fc_values[Anum_pg_publication_oid - 1] = ObjectIdGetDatum(fc_puboid);
	fc_values[Anum_pg_publication_puballtables - 1] =
		BoolGetDatum(fc_stmt->for_all_tables);
	fc_values[Anum_pg_publication_pubinsert - 1] =
		BoolGetDatum(fc_pubactions.pubinsert);
	fc_values[Anum_pg_publication_pubupdate - 1] =
		BoolGetDatum(fc_pubactions.pubupdate);
	fc_values[Anum_pg_publication_pubdelete - 1] =
		BoolGetDatum(fc_pubactions.pubdelete);
	fc_values[Anum_pg_publication_pubtruncate - 1] =
		BoolGetDatum(fc_pubactions.pubtruncate);
	fc_values[Anum_pg_publication_pubviaroot - 1] =
		BoolGetDatum(fc_publish_via_partition_root);

	fc_tup = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);

	/* 将元组插入目录。 */
	CatalogTupleInsert(fc_rel, fc_tup);
	heap_freetuple(fc_tup);

	recordDependencyOnOwner(PublicationRelationId, fc_puboid, GetUserId());

	ObjectAddressSet(fc_myself, PublicationRelationId, fc_puboid);

	/* 使更改可见。 */
	CommandCounterIncrement();

	/* 将对象与发布关联。 */
	if (fc_stmt->for_all_tables)
	{
		/* 使 relcache 无效以重建发布信息。 */
		CacheInvalidateRelcacheAll();
	}
	else
	{
		fc_ObjectsInPublicationToOids(fc_stmt->pubobjects, fc_pstate, &fc_relations,
								   &fc_schemaidlist);

		/* 对于 SCHEMA 中的 TABLES 要求超级用户 */
		if (fc_schemaidlist != NIL && !superuser())
			ereport(ERROR,
					errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					errmsg("must be superuser to create FOR TABLES IN SCHEMA publication"));

		if (list_length(fc_relations) > 0)
		{
			List	   *fc_rels;

			fc_rels = fc_OpenTableList(fc_relations);
			fc_TransformPubWhereClauses(fc_rels, fc_pstate->p_sourcetext,
									 fc_publish_via_partition_root);

			fc_CheckPubRelationColumnList(fc_stmt->pubname, fc_rels,
									   fc_schemaidlist != NIL,
									   fc_publish_via_partition_root);

			fc_PublicationAddTables(fc_puboid, fc_rels, true, NULL);
			fc_CloseTableList(fc_rels);
		}

		if (list_length(fc_schemaidlist) > 0)
		{
			/*
			 * 模式锁定在发布创建之前保持，以防止并发模式删除。
			 */
			fc_LockSchemaList(fc_schemaidlist);
			fc_PublicationAddSchemas(fc_puboid, fc_schemaidlist, true, NULL);
		}
	}

	table_close(fc_rel, RowExclusiveLock);

	InvokeObjectPostCreateHook(PublicationRelationId, fc_puboid, 0);

	if (wal_level != WAL_LEVEL_LOGICAL)
		ereport(WARNING,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("wal_level is insufficient to publish logical changes"),
				 errhint("Set wal_level to \"logical\" before creating subscriptions.")));

	return fc_myself;
}

/*
 * 更改发布的选项。
 */
static void fc_AlterPublicationOptions(ParseState *fc_pstate, AlterPublicationStmt *fc_stmt,
						Relation fc_rel, HeapTuple fc_tup)
{
	bool		fc_nulls[Natts_pg_publication];
	bool		fc_replaces[Natts_pg_publication];
	Datum		fc_values[Natts_pg_publication];
	bool		fc_publish_given;
	PublicationActions fc_pubactions;
	bool		fc_publish_via_partition_root_given;
	bool		fc_publish_via_partition_root;
	ObjectAddress fc_obj;
	Form_pg_publication fc_pubform;
	List	   *fc_root_relids = NIL;
	ListCell   *fc_lc;

	fc_parse_publication_options(fc_pstate,
							  fc_stmt->options,
							  &fc_publish_given, &fc_pubactions,
							  &fc_publish_via_partition_root_given,
							  &fc_publish_via_partition_root);

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	/*
	 * 如果发布不通过根分区表发布更改，则将使用分区的行过滤器和列列表。因此在这种情况下不允许在分区表上使用 WHERE 子句和列列表。
	 */
	if (!fc_pubform->puballtables && fc_publish_via_partition_root_given &&
		!fc_publish_via_partition_root)
	{
		/*
		 * 锁定发布，这样其他人就无法对此进行任何操作。这
		 * 可以防止并发的更改，添加带有 WHERE
		 * 子句和/或列列表的分区表，在不通过根发布时我们不允许这样做。
		 */
		LockDatabaseObject(PublicationRelationId, fc_pubform->oid, 0,
						   AccessShareLock);

		fc_root_relids = GetPublicationRelations(fc_pubform->oid,
											  PUBLICATION_PART_ROOT);

		foreach(fc_lc, fc_root_relids)
		{
			Oid			fc_relid = lfirst_oid(fc_lc);
			HeapTuple	fc_rftuple;
			char		fc_relkind;
			char	   *fc_relname;
			bool		fc_has_rowfilter;
			bool		fc_has_collist;

			/*
			 * 注意：我们没有对关系加锁，因此要默默处理
			 * 缓存查找返回 NULL 的情况。
			 */

			fc_rftuple = SearchSysCache2(PUBLICATIONRELMAP,
									  ObjectIdGetDatum(fc_relid),
									  ObjectIdGetDatum(fc_pubform->oid));
			if (!HeapTupleIsValid(fc_rftuple))
				continue;
			fc_has_rowfilter = !heap_attisnull(fc_rftuple, Anum_pg_publication_rel_prqual, NULL);
			fc_has_collist = !heap_attisnull(fc_rftuple, Anum_pg_publication_rel_prattrs, NULL);
			if (!fc_has_rowfilter && !fc_has_collist)
			{
				ReleaseSysCache(fc_rftuple);
				continue;
			}

			fc_relkind = get_rel_relkind(fc_relid);
			if (fc_relkind != RELKIND_PARTITIONED_TABLE)
			{
				ReleaseSysCache(fc_rftuple);
				continue;
			}
			fc_relname = get_rel_name(fc_relid);
			if (fc_relname == NULL)	/* 表同时被删除 */
			{
				ReleaseSysCache(fc_rftuple);
				continue;
			}

			if (fc_has_rowfilter)
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
								"publish_via_partition_root",
								fc_stmt->pubname),
						 errdetail("The publication contains a WHERE clause for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
								   fc_relname, "publish_via_partition_root")));
			Assert(fc_has_collist);
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
							"publish_via_partition_root",
							fc_stmt->pubname),
					 errdetail("The publication contains a column list for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
							   fc_relname, "publish_via_partition_root")));
		}
	}

	/* 一切正常，形成一个新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	memset(fc_replaces, false, sizeof(fc_replaces));

	if (fc_publish_given)
	{
		fc_values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(fc_pubactions.pubinsert);
		fc_replaces[Anum_pg_publication_pubinsert - 1] = true;

		fc_values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(fc_pubactions.pubupdate);
		fc_replaces[Anum_pg_publication_pubupdate - 1] = true;

		fc_values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(fc_pubactions.pubdelete);
		fc_replaces[Anum_pg_publication_pubdelete - 1] = true;

		fc_values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(fc_pubactions.pubtruncate);
		fc_replaces[Anum_pg_publication_pubtruncate - 1] = true;
	}

	if (fc_publish_via_partition_root_given)
	{
		fc_values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(fc_publish_via_partition_root);
		fc_replaces[Anum_pg_publication_pubviaroot - 1] = true;
	}

	fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_values, fc_nulls,
							fc_replaces);

	/* 更新目录。 */
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	CommandCounterIncrement();

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	/* 使 relcache 无效。 */
	if (fc_pubform->puballtables)
	{
		CacheInvalidateRelcacheAll();
	}
	else
	{
		List	   *fc_relids = NIL;
		List	   *fc_schemarelids = NIL;

		/*
		 * 对于包含在发布中的任何分区表，我们必须
		 * 使相应分区树中的所有分区无效，而不仅仅是那些在发布中明确提到的。
		 */
		if (fc_root_relids == NIL)
			fc_relids = GetPublicationRelations(fc_pubform->oid,
											 PUBLICATION_PART_ALL);
		else
		{
			/*
			 * 我们已经获得了在发布中明确提到的表。
			 * 现在获取列表中分区表的所有分区。
			 */
			foreach(fc_lc, fc_root_relids)
				fc_relids = GetPubPartitionOptionRelations(fc_relids,
														PUBLICATION_PART_ALL,
														lfirst_oid(fc_lc));
		}

		fc_schemarelids = GetAllSchemaPublicationRelations(fc_pubform->oid,
														PUBLICATION_PART_ALL);
		fc_relids = list_concat_unique_oid(fc_relids, fc_schemarelids);

		InvalidatePublicationRels(fc_relids);
	}

	ObjectAddressSet(fc_obj, PublicationRelationId, fc_pubform->oid);
	EventTriggerCollectSimpleCommand(fc_obj, InvalidObjectAddress,
									 (Node *) fc_stmt);

	InvokeObjectPostAlterHook(PublicationRelationId, fc_pubform->oid, 0);
}

/*
 * 使关系无效。
 */
void InvalidatePublicationRels(List *fc_relids)
{
	/*
	 * 我们不想发送太多单独的消息，到了某个时候
	 * 重新设置整个 relcache 更便宜。
	 */
	if (list_length(fc_relids) < MAX_RELCACHE_INVAL_MSGS)
	{
		ListCell   *fc_lc;

		foreach(fc_lc, fc_relids)
			CacheInvalidateRelcacheByRelid(lfirst_oid(fc_lc));
	}
	else
		CacheInvalidateRelcacheAll();
}

/*
 * 向发布中添加或从中移除表。
 */
static void fc_AlterPublicationTables(AlterPublicationStmt *fc_stmt, HeapTuple fc_tup,
					   List *fc_tables, const char *fc_queryString,
					   bool fc_publish_schema)
{
	List	   *fc_rels = NIL;
	Form_pg_publication fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);
	Oid			fc_pubid = fc_pubform->oid;

	/*
	 * 如果没有对象，则无需操作，除了在 SET 中：因为
	 * 用户可能没有指定任何表，在这种情况下我们需要
	 * 删除所有现有表。
	 */
	if (!fc_tables && fc_stmt->action != AP_SetObjects)
		return;

	fc_rels = fc_OpenTableList(fc_tables);

	if (fc_stmt->action == AP_AddObjects)
	{
		fc_TransformPubWhereClauses(fc_rels, fc_queryString, fc_pubform->pubviaroot);

		fc_publish_schema |= is_schema_publication(fc_pubid);

		fc_CheckPubRelationColumnList(fc_stmt->pubname, fc_rels, fc_publish_schema,
								   fc_pubform->pubviaroot);

		fc_PublicationAddTables(fc_pubid, fc_rels, false, fc_stmt);
	}
	else if (fc_stmt->action == AP_DropObjects)
		fc_PublicationDropTables(fc_pubid, fc_rels, false);
	else						/* AP_SetObjects */
	{
		List	   *fc_oldrelids = GetPublicationRelations(fc_pubid,
														PUBLICATION_PART_ROOT);
		List	   *fc_delrels = NIL;
		ListCell   *fc_oldlc;

		fc_TransformPubWhereClauses(fc_rels, fc_queryString, fc_pubform->pubviaroot);

		fc_CheckPubRelationColumnList(fc_stmt->pubname, fc_rels, fc_publish_schema,
								   fc_pubform->pubviaroot);

		/*
		 * 要重新创建发布的关系列表，查找
		 * 不需要被删除的现有关系。
		 */
		foreach(fc_oldlc, fc_oldrelids)
		{
			Oid			fc_oldrelid = lfirst_oid(fc_oldlc);
			ListCell   *fc_newlc;
			PublicationRelInfo *fc_oldrel;
			bool		fc_found = false;
			HeapTuple	fc_rftuple;
			Node	   *fc_oldrelwhereclause = NULL;
			Bitmapset  *fc_oldcolumns = NULL;

			/* 查找旧 relmap 的缓存 */
			fc_rftuple = SearchSysCache2(PUBLICATIONRELMAP,
									  ObjectIdGetDatum(fc_oldrelid),
									  ObjectIdGetDatum(fc_pubid));

			/*
			 * 查看现有关系当前是否有 WHERE 子句或
			 * 列列表。我们也需要比较这些。
			 */
			if (HeapTupleIsValid(fc_rftuple))
			{
				bool		fc_isnull = true;
				Datum		fc_whereClauseDatum;
				Datum		fc_columnListDatum;

				/* 加载此表的 WHERE 子句。 */
				fc_whereClauseDatum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_rftuple,
												   Anum_pg_publication_rel_prqual,
												   &fc_isnull);
				if (!fc_isnull)
					fc_oldrelwhereclause = stringToNode(TextDatumGetCString(fc_whereClauseDatum));

				/* 将 int2vector 列表转换为位图。 */
				fc_columnListDatum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_rftuple,
												  Anum_pg_publication_rel_prattrs,
												  &fc_isnull);

				if (!fc_isnull)
					fc_oldcolumns = pub_collist_to_bitmapset(NULL, fc_columnListDatum, NULL);

				ReleaseSysCache(fc_rftuple);
			}

			foreach(fc_newlc, fc_rels)
			{
				PublicationRelInfo *fc_newpubrel;
				Oid			fc_newrelid;
				Bitmapset  *fc_newcolumns = NULL;

				fc_newpubrel = (PublicationRelInfo *) lfirst(fc_newlc);
				fc_newrelid = RelationGetRelid(fc_newpubrel->relation);

				/*
				 * 如果新的发布有列列表，也将其转换为位图。
				 */
				if (fc_newpubrel->columns)
				{
					ListCell   *fc_lc;

					foreach(fc_lc, fc_newpubrel->columns)
					{
						char	   *fc_colname = strVal(lfirst(fc_lc));
						AttrNumber	fc_attnum = get_attnum(fc_newrelid, fc_colname);

						fc_newcolumns = bms_add_member(fc_newcolumns, fc_attnum);
					}
				}

				/*
				 * 检查新一组关系是否与
				 * 发布中的现有关系匹配。此外，如果
				 * 关系有一个关联的 WHERE 子句，请检查 WHERE
				 * 表达式是否也匹配。列列表同样如此。删除其余的。
				 */
				if (RelationGetRelid(fc_newpubrel->relation) == fc_oldrelid)
				{
					if (equal(fc_oldrelwhereclause, fc_newpubrel->whereClause) &&
						bms_equal(fc_oldcolumns, fc_newcolumns))
					{
						fc_found = true;
						break;
					}
				}
			}

			/*
			 * 将不匹配的关系添加到列表中，以便可以
			 * 删除它们。
			 */
			if (!fc_found)
			{
				fc_oldrel = palloc(sizeof(PublicationRelInfo));
				fc_oldrel->whereClause = NULL;
				fc_oldrel->columns = NIL;
				fc_oldrel->relation = table_open(fc_oldrelid,
											  ShareUpdateExclusiveLock);
				fc_delrels = lappend(fc_delrels, fc_oldrel);
			}
		}

		/* 并删除它们。 */
		fc_PublicationDropTables(fc_pubid, fc_delrels, true);

		/*
		 * 对于添加操作，不用费心计算差异，我们将在
		 * 更新目录时捕获并跳过现有的。
		 */
		fc_PublicationAddTables(fc_pubid, fc_rels, true, fc_stmt);

		fc_CloseTableList(fc_delrels);
	}

	fc_CloseTableList(fc_rels);
}

/*
 * 更改发布模式。
 *
 * 向发布中添加或从中移除模式。
 */
static void fc_AlterPublicationSchemas(AlterPublicationStmt *fc_stmt,
						HeapTuple fc_tup, List *fc_schemaidlist)
{
	Form_pg_publication fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	/*
	 * 如果没有对象，则无需操作，除了在 SET 中：因为
	 * 用户可能没有指定任何模式，在这种情况下我们需要
	 * 删除所有现有模式。
	 */
	if (!fc_schemaidlist && fc_stmt->action != AP_SetObjects)
		return;

	/*
	 * 在更改发布时保持模式锁定，以防止
	 * 并发的模式删除。
	 */
	fc_LockSchemaList(fc_schemaidlist);
	if (fc_stmt->action == AP_AddObjects)
	{
		ListCell   *fc_lc;
		List	   *fc_reloids;

		fc_reloids = GetPublicationRelations(fc_pubform->oid, PUBLICATION_PART_ROOT);

		foreach(fc_lc, fc_reloids)
		{
			HeapTuple	fc_coltuple;

			fc_coltuple = SearchSysCache2(PUBLICATIONRELMAP,
									   ObjectIdGetDatum(lfirst_oid(fc_lc)),
									   ObjectIdGetDatum(fc_pubform->oid));

			if (!HeapTupleIsValid(fc_coltuple))
				continue;

			/*
			 * 如果列列表已经是发布的一部分，则不允许添加模式。
			 * 请参见 CheckPubRelationColumnList。
			 */
			if (!heap_attisnull(fc_coltuple, Anum_pg_publication_rel_prattrs, NULL))
				ereport(ERROR,
						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						errmsg("cannot add schema to publication \"%s\"",
							   fc_stmt->pubname),
						errdetail("Schemas cannot be added if any tables that specify a column list are already part of the publication."));

			ReleaseSysCache(fc_coltuple);
		}

		fc_PublicationAddSchemas(fc_pubform->oid, fc_schemaidlist, false, fc_stmt);
	}
	else if (fc_stmt->action == AP_DropObjects)
		fc_PublicationDropSchemas(fc_pubform->oid, fc_schemaidlist, false);
	else						/* AP_SetObjects */
	{
		List	   *fc_oldschemaids = GetPublicationSchemas(fc_pubform->oid);
		List	   *fc_delschemas = NIL;

		/* 确定应该删除哪些模式 */
		fc_delschemas = list_difference_oid(fc_oldschemaids, fc_schemaidlist);

		/*
		 * 模式锁在发布被更改之前保持，以防止
		 * 并行模式删除。
		 */
		fc_LockSchemaList(fc_delschemas);

		/* 并删除它们 */
		fc_PublicationDropSchemas(fc_pubform->oid, fc_delschemas, true);

		/*
		 * 对于添加操作，不用费心计算差异，我们将在
		 * 更新目录时捕获并跳过现有的。
		 */
		fc_PublicationAddSchemas(fc_pubform->oid, fc_schemaidlist, true, fc_stmt);
	}
}

/*
 * 检查关系和模式是否可以在给定的发布中，并在不可以时抛出
 * 适当的错误。
 */
static void fc_CheckAlterPublication(AlterPublicationStmt *fc_stmt, HeapTuple fc_tup,
					  List *fc_tables, List *fc_schemaidlist)
{
	Form_pg_publication fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	if ((fc_stmt->action == AP_AddObjects || fc_stmt->action == AP_SetObjects) &&
		fc_schemaidlist && !superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to add or set schemas")));

	/*
	 * 检查用户是否被允许操作模式中的发布表
	 */
	if (fc_schemaidlist && fc_pubform->puballtables)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("publication \"%s\" is defined as FOR ALL TABLES",
						NameStr(fc_pubform->pubname)),
				 errdetail("Schemas cannot be added to or dropped from FOR ALL TABLES publications.")));

	/* 检查用户是否被允许操作发布表。 */
	if (fc_tables && fc_pubform->puballtables)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("publication \"%s\" is defined as FOR ALL TABLES",
						NameStr(fc_pubform->pubname)),
				 errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications.")));
}

/*
 * 更改现有的发布。
 *
 * 这是AlterPublicationOptions、AlterPublicationSchemas和AlterPublicationTables的调度函数。
 */
void AlterPublication(ParseState *fc_pstate, AlterPublicationStmt *fc_stmt)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	Form_pg_publication fc_pubform;

	fc_rel = table_open(PublicationRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy1(PUBLICATIONNAME,
							  CStringGetDatum(fc_stmt->pubname));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("publication \"%s\" does not exist",
						fc_stmt->pubname)));

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	/* 必须是所有者 */
	if (!pg_publication_ownercheck(fc_pubform->oid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
					   fc_stmt->pubname);

	if (fc_stmt->options)
		fc_AlterPublicationOptions(fc_pstate, fc_stmt, fc_rel, fc_tup);
	else
	{
		List	   *fc_relations = NIL;
		List	   *fc_schemaidlist = NIL;
		Oid			fc_pubid = fc_pubform->oid;

		fc_ObjectsInPublicationToOids(fc_stmt->pubobjects, fc_pstate, &fc_relations,
								   &fc_schemaidlist);

		fc_CheckAlterPublication(fc_stmt, fc_tup, fc_relations, fc_schemaidlist);

		heap_freetuple(fc_tup);

		/* 锁定发布，以便没有其他人可以对此进行任何操作。 */
		LockDatabaseObject(PublicationRelationId, fc_pubid, 0,
						   AccessExclusiveLock);

		/*
		 * 在我们获得发布锁之前，可能已经有并行DDL将其删除。
		 * 我们可以通过检查发布的存在性来测试这一点。
		 * 我们再次获取元组以避免任何发布选项被更改的风险。
		 */
		fc_tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(fc_pubid));
		if (!HeapTupleIsValid(fc_tup))
			ereport(ERROR,
					errcode(ERRCODE_UNDEFINED_OBJECT),
					errmsg("publication \"%s\" does not exist",
						   fc_stmt->pubname));

		fc_AlterPublicationTables(fc_stmt, fc_tup, fc_relations, fc_pstate->p_sourcetext,
							   fc_schemaidlist != NIL);
		fc_AlterPublicationSchemas(fc_stmt, fc_tup, fc_schemaidlist);
	}

	/* 清理。 */
	heap_freetuple(fc_tup);
	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 通过映射OID从发布中移除关系。
 */
void RemovePublicationRelById(Oid fc_proid)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	Form_pg_publication_rel fc_pubrel;
	List	   *fc_relids = NIL;

	fc_rel = table_open(PublicationRelRelationId, RowExclusiveLock);

	fc_tup = SearchSysCache1(PUBLICATIONREL, ObjectIdGetDatum(fc_proid));

	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for publication table %u",
			 fc_proid);

	fc_pubrel = (Form_pg_publication_rel) GETSTRUCT(fc_tup);

	/*
	 * 使relcache失效，以便重建发布信息。
	 *
	 * 对于分区表，我们必须使所包含的所有分区失效
	 * 在各自的分区层次结构中，而不仅仅是显式
	 * 在发布中提到的那个。这是必需的，因为当发布父表时，
	 * 我们隐式地发布子表。
	 */
	fc_relids = GetPubPartitionOptionRelations(fc_relids, PUBLICATION_PART_ALL,
											fc_pubrel->prrelid);

	InvalidatePublicationRels(fc_relids);

	CatalogTupleDelete(fc_rel, &fc_tup->t_self);

	ReleaseSysCache(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 通过映射OID移除发布。
 */
void RemovePublicationById(Oid fc_pubid)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	Form_pg_publication fc_pubform;

	fc_rel = table_open(PublicationRelationId, RowExclusiveLock);

	fc_tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(fc_pubid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for publication %u", fc_pubid);

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	/* 使 relcache 无效以重建发布信息。 */
	if (fc_pubform->puballtables)
		CacheInvalidateRelcacheAll();

	CatalogTupleDelete(fc_rel, &fc_tup->t_self);

	ReleaseSysCache(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 通过映射OID从发布中移除模式。
 */
void RemovePublicationSchemaById(Oid fc_psoid)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	List	   *fc_schemaRels = NIL;
	Form_pg_publication_namespace fc_pubsch;

	fc_rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);

	fc_tup = SearchSysCache1(PUBLICATIONNAMESPACE, ObjectIdGetDatum(fc_psoid));

	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for publication schema %u", fc_psoid);

	fc_pubsch = (Form_pg_publication_namespace) GETSTRUCT(fc_tup);

	/*
	 * 使relcache失效，以便重建发布信息。请参见
	 * RemovePublicationRelById，了解我们为何需要考虑所有
	 * 分区。
	 */
	fc_schemaRels = GetSchemaPublicationRelations(fc_pubsch->pnnspid,
											   PUBLICATION_PART_ALL);
	InvalidatePublicationRels(fc_schemaRels);

	CatalogTupleDelete(fc_rel, &fc_tup->t_self);

	ReleaseSysCache(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 打开由PublicationTable列表指定的关系。
 * 返回的表以ShareUpdateExclusiveLock模式锁定，以便
 * 将它们添加到发布中。
 */
static List * fc_OpenTableList(List *fc_tables)
{
	List	   *fc_relids = NIL;
	List	   *fc_rels = NIL;
	ListCell   *fc_lc;
	List	   *fc_relids_with_rf = NIL;
	List	   *fc_relids_with_collist = NIL;

	/*
	 * 打开、共享锁定并检查所有显式指定的关系
	 */
	foreach(fc_lc, fc_tables)
	{
		PublicationTable *fc_t = lfirst_node(PublicationTable, fc_lc);
		bool		fc_recurse = fc_t->relation->inh;
		Relation	fc_rel;
		Oid			fc_myrelid;
		PublicationRelInfo *fc_pub_rel;

		/* 允许查询取消，以防这耗时较长 */
		CHECK_FOR_INTERRUPTS();

		fc_rel = table_openrv(fc_t->relation, ShareUpdateExclusiveLock);
		fc_myrelid = RelationGetRelid(fc_rel);

		/*
		 * 如果用户指定“foo, foo”，请过滤掉重复项。
		 *
		 * 请注意，这种算法在效率上并不高效（O(N^2)），
		 * 但考虑到它仅适用于用户给出的表列表，
		 * 因此被认为是可以接受的。
		 */
		if (list_member_oid(fc_relids, fc_myrelid))
		{
			/* 如果有任何带行过滤器的表，则不允许重复表。 */
			if (fc_t->whereClause || list_member_oid(fc_relids_with_rf, fc_myrelid))
				ereport(ERROR,
						(errcode(ERRCODE_DUPLICATE_OBJECT),
						 errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
								RelationGetRelationName(fc_rel))));

			/* 如果有任何带列列表的表，则不允许重复表。 */
			if (fc_t->columns || list_member_oid(fc_relids_with_collist, fc_myrelid))
				ereport(ERROR,
						(errcode(ERRCODE_DUPLICATE_OBJECT),
						 errmsg("conflicting or redundant column lists for table \"%s\"",
								RelationGetRelationName(fc_rel))));

			table_close(fc_rel, ShareUpdateExclusiveLock);
			continue;
		}

		fc_pub_rel = palloc(sizeof(PublicationRelInfo));
		fc_pub_rel->relation = fc_rel;
		fc_pub_rel->whereClause = fc_t->whereClause;
		fc_pub_rel->columns = fc_t->columns;
		fc_rels = lappend(fc_rels, fc_pub_rel);
		fc_relids = lappend_oid(fc_relids, fc_myrelid);

		if (fc_t->whereClause)
			fc_relids_with_rf = lappend_oid(fc_relids_with_rf, fc_myrelid);

		if (fc_t->columns)
			fc_relids_with_collist = lappend_oid(fc_relids_with_collist, fc_myrelid);

		/*
		 * 如果请求，添加此关系的子项，以便它们也被添加
		 * 到发布中。分区表不能有任何继承子项， 
		 * 除了其分区，分区不需要显式
		 * 添加到发布中。
		 */
		if (fc_recurse && fc_rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
		{
			List	   *fc_children;
			ListCell   *fc_child;

			fc_children = find_all_inheritors(fc_myrelid, ShareUpdateExclusiveLock,
										   NULL);

			foreach(fc_child, fc_children)
			{
				Oid			fc_childrelid = lfirst_oid(fc_child);

				/* 允许查询取消，以防这耗时较长 */
				CHECK_FOR_INTERRUPTS();

				/*
				 * 如果用户同时指定了父表和子表，则跳过重复项。
				 */
				if (list_member_oid(fc_relids, fc_childrelid))
				{
					/*
					 * 我们不允许同时为父表和子表指定行过滤器，因为不太清楚应优先考虑哪一个。
					 */
					if (fc_childrelid != fc_myrelid &&
						(fc_t->whereClause || list_member_oid(fc_relids_with_rf, fc_childrelid)))
						ereport(ERROR,
								(errcode(ERRCODE_DUPLICATE_OBJECT),
								 errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
										RelationGetRelationName(fc_rel))));

					/*
					 * 我们不允许同时为父表和子表指定列列表，因为不太清楚应优先考虑哪一个。
					 */
					if (fc_childrelid != fc_myrelid &&
						(fc_t->columns || list_member_oid(fc_relids_with_collist, fc_childrelid)))
						ereport(ERROR,
								(errcode(ERRCODE_DUPLICATE_OBJECT),
								 errmsg("conflicting or redundant column lists for table \"%s\"",
										RelationGetRelationName(fc_rel))));

					continue;
				}

				/* find_all_inheritors 已经获得锁 */
				fc_rel = table_open(fc_childrelid, NoLock);
				fc_pub_rel = palloc(sizeof(PublicationRelInfo));
				fc_pub_rel->relation = fc_rel;
				/* 子表从父表继承 WHERE 子句 */
				fc_pub_rel->whereClause = fc_t->whereClause;

				/* 子表从父表继承列列表 */
				fc_pub_rel->columns = fc_t->columns;
				fc_rels = lappend(fc_rels, fc_pub_rel);
				fc_relids = lappend_oid(fc_relids, fc_childrelid);

				if (fc_t->whereClause)
					fc_relids_with_rf = lappend_oid(fc_relids_with_rf, fc_childrelid);

				if (fc_t->columns)
					fc_relids_with_collist = lappend_oid(fc_relids_with_collist, fc_childrelid);
			}
		}
	}

	list_free(fc_relids);
	list_free(fc_relids_with_rf);

	return fc_rels;
}

/*
 * 关闭列表中的所有关系。
 */
static void fc_CloseTableList(List *fc_rels)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_rels)
	{
		PublicationRelInfo *fc_pub_rel;

		fc_pub_rel = (PublicationRelInfo *) lfirst(fc_lc);
		table_close(fc_pub_rel->relation, NoLock);
	}

	list_free_deep(fc_rels);
}

/*
 * 在 AccessShareLock 模式下锁定架构列表中指定的架构，以防止并发架构删除。
 */
static void fc_LockSchemaList(List *fc_schemalist)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_schemalist)
	{
		Oid			fc_schemaid = lfirst_oid(fc_lc);

		/* 允许查询取消，以防这耗时较长 */
		CHECK_FOR_INTERRUPTS();
		LockDatabaseObject(NamespaceRelationId, fc_schemaid, 0, AccessShareLock);

		/*
		 * 有可能在我们获得架构锁时，并发的 DDL 已将其删除。我们可以通过检查架构的存在性来测试这一点。
		 */
		if (!SearchSysCacheExists1(NAMESPACEOID, ObjectIdGetDatum(fc_schemaid)))
			ereport(ERROR,
					errcode(ERRCODE_UNDEFINED_SCHEMA),
					errmsg("schema with OID %u does not exist", fc_schemaid));
	}
}

/*
 * 将列出的表添加到发布中。
 */
static void fc_PublicationAddTables(Oid fc_pubid, List *fc_rels, bool fc_if_not_exists,
					 AlterPublicationStmt *fc_stmt)
{
	ListCell   *fc_lc;

	Assert(!fc_stmt || !fc_stmt->for_all_tables);

	foreach(fc_lc, fc_rels)
	{
		PublicationRelInfo *fc_pub_rel = (PublicationRelInfo *) lfirst(fc_lc);
		Relation	fc_rel = fc_pub_rel->relation;
		ObjectAddress fc_obj;

		/* 必须是表的拥有者或超级用户。 */
		if (!pg_class_ownercheck(RelationGetRelid(fc_rel), GetUserId()))
			aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(fc_rel->rd_rel->relkind),
						   RelationGetRelationName(fc_rel));

		fc_obj = publication_add_relation(fc_pubid, fc_pub_rel, fc_if_not_exists);
		if (fc_stmt)
		{
			EventTriggerCollectSimpleCommand(fc_obj, InvalidObjectAddress,
											 (Node *) fc_stmt);

			InvokeObjectPostCreateHook(PublicationRelRelationId,
									   fc_obj.objectId, 0);
		}
	}
}

/*
 * 从发布中移除列出的表。
 */
static void fc_PublicationDropTables(Oid fc_pubid, List *fc_rels, bool fc_missing_ok)
{
	ObjectAddress fc_obj;
	ListCell   *fc_lc;
	Oid			fc_prid;

	foreach(fc_lc, fc_rels)
	{
		PublicationRelInfo *fc_pubrel = (PublicationRelInfo *) lfirst(fc_lc);
		Relation	fc_rel = fc_pubrel->relation;
		Oid			fc_relid = RelationGetRelid(fc_rel);

		if (fc_pubrel->columns)
			ereport(ERROR,
					errcode(ERRCODE_SYNTAX_ERROR),
					errmsg("column list must not be specified in ALTER PUBLICATION ... DROP"));

		fc_prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid,
							   ObjectIdGetDatum(fc_relid),
							   ObjectIdGetDatum(fc_pubid));
		if (!OidIsValid(fc_prid))
		{
			if (fc_missing_ok)
				continue;

			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("relation \"%s\" is not part of the publication",
							RelationGetRelationName(fc_rel))));
		}

		if (fc_pubrel->whereClause)
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("cannot use a WHERE clause when removing a table from a publication")));

		ObjectAddressSet(fc_obj, PublicationRelRelationId, fc_prid);
		performDeletion(&fc_obj, DROP_CASCADE, 0);
	}
}

/*
 * 将列出的架构添加到发布中。
 */
static void fc_PublicationAddSchemas(Oid fc_pubid, List *fc_schemas, bool fc_if_not_exists,
					  AlterPublicationStmt *fc_stmt)
{
	ListCell   *fc_lc;

	Assert(!fc_stmt || !fc_stmt->for_all_tables);

	foreach(fc_lc, fc_schemas)
	{
		Oid			fc_schemaid = lfirst_oid(fc_lc);
		ObjectAddress fc_obj;

		fc_obj = publication_add_schema(fc_pubid, fc_schemaid, fc_if_not_exists);
		if (fc_stmt)
		{
			EventTriggerCollectSimpleCommand(fc_obj, InvalidObjectAddress,
											 (Node *) fc_stmt);

			InvokeObjectPostCreateHook(PublicationNamespaceRelationId,
									   fc_obj.objectId, 0);
		}
	}
}

/*
 * 从发布中移除列出的架构。
 */
static void fc_PublicationDropSchemas(Oid fc_pubid, List *fc_schemas, bool fc_missing_ok)
{
	ObjectAddress fc_obj;
	ListCell   *fc_lc;
	Oid			fc_psid;

	foreach(fc_lc, fc_schemas)
	{
		Oid			fc_schemaid = lfirst_oid(fc_lc);

		fc_psid = GetSysCacheOid2(PUBLICATIONNAMESPACEMAP,
							   Anum_pg_publication_namespace_oid,
							   ObjectIdGetDatum(fc_schemaid),
							   ObjectIdGetDatum(fc_pubid));
		if (!OidIsValid(fc_psid))
		{
			if (fc_missing_ok)
				continue;

			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("tables from schema \"%s\" are not part of the publication",
							get_namespace_name(fc_schemaid))));
		}

		ObjectAddressSet(fc_obj, PublicationNamespaceRelationId, fc_psid);
		performDeletion(&fc_obj, DROP_CASCADE, 0);
	}
}

/*
 * 用于更改发布拥有者的内部工作工具
 */
static void fc_AlterPublicationOwner_internal(Relation fc_rel, HeapTuple fc_tup, Oid fc_newOwnerId)
{
	Form_pg_publication fc_form;

	fc_form = (Form_pg_publication) GETSTRUCT(fc_tup);

	if (fc_form->pubowner == fc_newOwnerId)
		return;

	if (!superuser())
	{
		AclResult	fc_aclresult;

		/* 必须是拥有者 */
		if (!pg_publication_ownercheck(fc_form->oid, GetUserId()))
			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
						   NameStr(fc_form->pubname));

		/* 必须能够成为新所有者 */
		check_is_member_of_role(GetUserId(), fc_newOwnerId);

		/* 新拥有者必须在数据库上具有 CREATE 权限 */
		fc_aclresult = pg_database_aclcheck(MyDatabaseId, fc_newOwnerId, ACL_CREATE);
		if (fc_aclresult != ACLCHECK_OK)
			aclcheck_error(fc_aclresult, OBJECT_DATABASE,
						   get_database_name(MyDatabaseId));

		if (fc_form->puballtables && !superuser_arg(fc_newOwnerId))
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied to change owner of publication \"%s\"",
							NameStr(fc_form->pubname)),
					 errhint("The owner of a FOR ALL TABLES publication must be a superuser.")));

		if (!superuser_arg(fc_newOwnerId) && is_schema_publication(fc_form->oid))
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied to change owner of publication \"%s\"",
							NameStr(fc_form->pubname)),
					 errhint("The owner of a FOR TABLES IN SCHEMA publication must be a superuser.")));
	}

	fc_form->pubowner = fc_newOwnerId;
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	/* 更新所有者依赖关系引用 */
	changeDependencyOnOwner(PublicationRelationId,
							fc_form->oid,
							fc_newOwnerId);

	InvokeObjectPostAlterHook(PublicationRelationId,
							  fc_form->oid, 0);
}

/*
 * 通过名称更改发布拥有者
 */
ObjectAddress AlterPublicationOwner(const char *fc_name, Oid fc_newOwnerId)
{
	Oid			fc_subid;
	HeapTuple	fc_tup;
	Relation	fc_rel;
	ObjectAddress fc_address;
	Form_pg_publication fc_pubform;

	fc_rel = table_open(PublicationRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy1(PUBLICATIONNAME, CStringGetDatum(fc_name));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("publication \"%s\" does not exist", fc_name)));

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);
	fc_subid = fc_pubform->oid;

	fc_AlterPublicationOwner_internal(fc_rel, fc_tup, fc_newOwnerId);

	ObjectAddressSet(fc_address, PublicationRelationId, fc_subid);

	heap_freetuple(fc_tup);

	table_close(fc_rel, RowExclusiveLock);

	return fc_address;
}

/*
 * 通过 OID 更改发布拥有者
 */
void AlterPublicationOwner_oid(Oid fc_subid, Oid fc_newOwnerId)
{
	HeapTuple	fc_tup;
	Relation	fc_rel;

	fc_rel = table_open(PublicationRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("publication with OID %u does not exist", fc_subid)));

	fc_AlterPublicationOwner_internal(fc_rel, fc_tup, fc_newOwnerId);

	heap_freetuple(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}
